#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "ringqueue.h"
#include "log.h"
#include "threadpool.h"

// 初始队列大小
#define TP_QUEUE_SIZE 32

typedef struct {
    pthread_t tid;
    unsigned char terminate;    // 是否已终止
    struct threadpool *pool;    // 指向线程池
} thread_t;

struct threadpool {
    rqueue_t *queue;              // task queue
    int num_thread, num_active;         // num threads
    int thread_quit;            // flag indicate thread should quit
    thread_t *thread;          // thread id array
};

// 线程任务：是队列的项，由工作线程处理
typedef struct {
    thread_task_func func;
    void *priv_data;
} thread_task_t;


void *thread_worker(void *arg)
{
    thread_task_t *task = NULL;
    thread_t *thread = (thread_t *)arg;
    threadpool_t tp = thread->pool;

    // 活动线程数+1
    tp->num_active += 1;

    while (!tp->thread_quit) {
        if ((task = rqueue_get(tp->queue, 1000))) {
            log_err("thread pool dequeue failed\n");
            break;
        }
        log_debug("thread %d executing...\n", (int)thread->tid);
        if (task) {
            // 执行线程任务
            task->func(task->priv_data);
            free(task);
        }
    }

    // 活动线程减1
    tp->num_active -= 1;
    thread->terminate = 1;
    log_info("thread worker quit\n");
    return (void *)0;
}


threadpool_t thread_pool_create(int num_thread)
{
    int i;
    pthread_attr_t attr;
    threadpool_t tp = NULL;
    thread_t *threadp = NULL;

    tp = calloc(1, sizeof(struct threadpool));
    if (!tp) {
        goto err_return;
    }
    if (num_thread > TP_MAX_THREADS)
        num_thread = TP_MAX_THREADS;

    // 初始化队列
    tp->queue = rqueue_init(TP_QUEUE_SIZE);
    if (!tp->queue) {
        goto err_return;
    }

    tp->num_thread = num_thread;
    tp->num_active= 0;
    tp->thread_quit = 0;

    // 创建工作线程
    tp->thread = calloc(num_thread, sizeof(thread_t));
    if (!tp->thread) {
        goto err_return;
    }
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); // detach模式
    for (i=0, threadp=tp->thread; i<num_thread; i++, threadp++) {
        threadp->terminate = 0;
        threadp->pool = tp;
        pthread_create(&threadp->tid, &attr, thread_worker, threadp);
    }
    pthread_attr_destroy(&attr);

    return tp;

err_return:
    if (tp->thread)
        free(tp->thread);
    if (tp)    
        free(tp);

    return NULL;    
}

int thread_pool_add_task(threadpool_t tp, thread_task_func func, void *priv)
{
    int ret;
    thread_task_t *task = NULL;

    task = calloc(1, sizeof(thread_task_t));
    if (!task) {
        return 0;
    }

    // 构造task
    task->func = func;
    task->priv_data = priv;

    // 放入队列
    ret = rqueue_put(tp->queue, task, 500);
    if (ret) {
        log_err("thread pool enqueue failed: %s\n", rqueue_strerror(ret));
    }
    return ret == 0;
}

/*
 * 模拟退出任务，用于工作线程退出
 */
static int enable_thread_quit(void *priv)
{
    threadpool_t tp = (threadpool_t)priv;
    tp->thread_quit = 1;

    return 1;
}

int thread_pool_destroy(threadpool_t tp)
{
    int i;

    /*
     * 工作线程的退出机制
     * 在线程池销毁时，发送与线程数相同数量的task
     * 此task设置thread_quit标志为1，目的是唤醒阻塞在读操作的线程，保证已detach线程最终退出
     */
    for (i=0; i<tp->num_thread; i++) {
        thread_pool_add_task(tp, enable_thread_quit, tp);
    }

    // 由于工作线程处于detach状态，实际上不需考虑[模拟退出任务]执行的具体时间，我们只是好奇追踪一下
    log_debug("thread pool destroy, %d thread(s) in active\n", tp->num_active);
    for (i=0; i<tp->num_thread;) {
        if (!tp->thread[i].terminate) { // 保证工作线程已退出
            usleep(20 * 1000);
            log_debug("thread (%d) unterminated, wait...\n", i);
            continue;
        }
        log_info("thread (%d) terminated\n", i);
        i++;
    }
    log_info("thread pool destroy\n\n");
    rqueue_destroy(&tp->queue, 0);
    free(tp->thread);
    free(tp);

    return 1;
}

